package com.belf.console.agent.migrate.job;

import belf.migrate.engine.admin.taskconf.SinkConf;
import belf.migrate.engine.admin.taskconf.SourceConf;
import com.alibaba.fastjson2.JSONObject;
import com.belf.common.core.domain.AjaxResult;
import com.belf.console.agent.migrate.BuildTaskConf;
import com.belf.console.agent.migrate.JobPush;
import com.belf.console.domain.DataSourceInfo;
import com.belf.console.domain.DataTaskConf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * SqlExecute
 *
 * @author l
 * @since 2024/8/22 13:51
 */

public class ExecuteSqlTask {

    private static final Logger logger = LoggerFactory.getLogger(ExecuteSqlTask.class.getName());

    public static AjaxResult executeSql(String APIPath,DataTaskConf taskData, JSONObject params ) {

        try {
            List<DataSourceInfo> dataSourceInfoList = taskData.getDataSourceInfoList();
            if (dataSourceInfoList != null && dataSourceInfoList.size() > 0) {
                for (DataSourceInfo dataSourceInfo : dataSourceInfoList) {
                    if (dataSourceInfo.getId().equals(taskData.getSourceId())) {
                        SourceConf sourceConf = BuildTaskConf.buildSourceConf(dataSourceInfo);
                        params.put("source", sourceConf);
                    }
                    if (dataSourceInfo.getId().equals(taskData.getSinkId())) {
                        //  连接器
                        SinkConf sinkConf = BuildTaskConf.buildSinkConf(dataSourceInfo);
                        params.put("sink", sinkConf);
                    }
                }
            }
            return JobPush.pushTask(APIPath, params.toJSONString());
        } catch (Exception e) {
            logger.error("执行sql失败", e);
        }
        return AjaxResult.error();
    }
}
